package net.xo68.destiny.core.eventbus.impl;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import lombok.extern.slf4j.Slf4j;
import net.xo68.destiny.core.eventbus.*;

import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 事件巴士管理者实现
 * @author wuxie
 * @version 2020-11-13
 */
@Slf4j
public class EventBusManagerImpl implements EventBusManager {

    private final EventBusConfigurer eventBusConfigurer;
    private final int ringBufferSize;
    private final ThreadFactory threadFactory;
    private final EventHandleRouting eventHandleRouting;

    private Disruptor<MessageEvent> disruptor;

    private final AtomicBoolean started=new AtomicBoolean(false);


    public EventBusManagerImpl(EventBusConfigurer eventBusConfigurer, int ringBufferSize, ThreadFactory threadFactory) {
        this.eventBusConfigurer = eventBusConfigurer;
        this.ringBufferSize = ringBufferSize;
        this.threadFactory = threadFactory;
        this.eventHandleRouting=new EventHandleRouting();
    }

    @Override
    public void destroy() throws Exception {
        if(started.compareAndSet(true,false)){
            disruptor.shutdown();
            disruptor=null;
            log.info("disruptor shutdown");
        }
    }

    @Override
    public void afterSingletonsInstantiated() {
        if(started.compareAndSet(false,true)){
            disruptor=new Disruptor<>(new MessageEventFactory(),
                    ringBufferSize,
                    threadFactory,
                    ProducerType.SINGLE,
                    new BlockingWaitStrategy());

            //this.eventHandleConfig.config(this.disruptor);
            this.eventBusConfigurer.configure(this.eventHandleRouting);

            this.disruptor.handleEventsWith(this.eventHandleRouting);

            this.disruptor.setDefaultExceptionHandler(new MessageEventExceptionHandler());
            this.disruptor.handleExceptionsFor(new ExceptionEventHandle());
            this.disruptor.start();
        }
    }

    @Override
    public void publish(int eventType, Object message){
        long sequence = this.disruptor.getRingBuffer().next(); // Grab the next sequence //获取下一个可用的序列号
        try
        {
            MessageEvent event = this.disruptor.getRingBuffer().get(sequence);
            // Get the entry in the Disruptor //通过序列号获取空闲可用的LongEvent
            // for the sequence
            event.setEventType(eventType);
            event.setBody(message); // Fill with data //设置数值
        }
        finally
        {
            this.disruptor.getRingBuffer().publish(sequence); //数据发布，只有发布后的数据才会真正被消费者看见
        }


    }
}
